package com.qingguo.Test

import java.security.MessageDigest

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}

object MD5All {

  def main(args: Array[String]): Unit = {


    /**
      * (0,1@!@张三@!@23,30f3881278bb0512f12c1a8d26e0fe4e)
      * (1,2@!@李四@!@21,653259a1b2a1eb0b9d5697f937af40df)
      *
      * 对每行进行加密：10李六20——>75be6a4edc94d79696da39f50a0f6cbc
      *
      */


    val tableName = "S02_DEMAND_MANAGE"

    val count = 5

    val whereExp = " "

    val conf = new SparkConf().setMaster("local").setAppName("MD5")
    val sc = new SparkContext(conf)

    conf.set("spark.sql.shuffle.partitions", "10")

    // spark  sql 上下文对象  spark sql的入口
    val spark = new SQLContext(sc)

    //读取json文件创建df  ,json文件每一行必须是一个对象
    val hiveDF: DataFrame = spark.read.json("spark/data/hive.json")

    //注册成表S02_DEMAND_MANAGE
    hiveDF.registerTempTable(tableName)


    /**
      * 插入MD5的值
      */
    val md5 = new Md5Verifier()
    val sqlFullData = spark.sql("SELECT * from " + tableName + whereExp)
    val sqlData = sqlFullData

    //    sqlFullData.show()

    /**
      * +-----------+-----------+-----------+--------+---+--------------------+----------------+---------+
      * |demand_code|demand_desc|demand_name|      dt| id|          md5_coding|person_in_charge|work_load|
      * +-----------+-----------+-----------+--------+---+--------------------+----------------+---------+
      * |     需求号|    需求描述|    需求名称|20201001|  1|90bac8cc578017e2b...|        需求负责|        4|
      * +-----------+-----------+-----------+--------+---+--------------------+----------------+---------+
      *
      */

    val rowRdd = sqlData.rdd.map(row => {
      //fieldIndex取字段所对应的下标
      val dtIdx = row.fieldIndex("dt")
      val md5Idx = row.fieldIndex("md5_coding")
      //[Ljava.lang.Object;@26a01196
      //[Ljava.lang.Object;@30679321
      val rowDataArray = row.toSeq.toArray
      val rowDataSb = new StringBuilder

      for (i <- 0 until row.length) {
        val value = String.valueOf(row.get(i))
        if (i != dtIdx && i != md5Idx) {
          if (!"null".equals(value)) {
            rowDataSb.append(md5.replaceDouble(value))
          }

          /** ceshi,id
            * 11
            * 12
            * 13
            * 14
            */
        }
      }
      //加密
      /**
        * 6512bd43d9caa6e02c990b0a82652dca
        * c20ad4d76fe97759aa27a0c99bff6710
        * c51ce410c124a10e0db5e4b97fc2af39
        */

//      println(rowDataSb)
      //需求号需求描述需求名称1需求负责4
      //fa68bce8e859485f327504bbcd05010d

      val md5Val = md5.md5Verify(rowDataSb.toString())
      rowDataArray(md5Idx) = md5Val

//      val newRow = row.fromSeq(rowDataArray)

      rowDataArray(md5Idx)
    })

//    rowRdd.foreach(println)

    //    spark.createDataFrame(rowRdd,sqlData.schema).write.mode("overwrite").insertInto(tableName)

  }


  //
  //    //    [age: bigint, id: string, md5_coding: string]
  //    //    val sqlFullData = sqlContext.sql("SELECT * from " + tableName + whereExp)
  //
  //    //res:[md5_coding: string]
  //    val sqlFullData: DataFrame = spark.sql("SELECT * from " + tableName + whereExp).select("md5_coding")
  //
  //    sqlFullData.show()
  //
  //    val inFileRDD = spark.sparkContext.textFile("spark/data/Md5data")
  //
  //    val md5 = new Md5Verifier
  //
  //    val inFileMD5RDD = inFileRDD
  //      .zipWithIndex()
  //      .map(x => {
  //        //  mkString把集合转成字符串
  //        //  data:1张三23为每一行的数据
  //        val data = x._1.split("@!@").map(s => md5.replaceDouble(s)).mkString
  //        //      取加密MD5的值
  //        val md5Val = md5.md5Verify(data)
  //        //        val md5Val = x._1.split("@!@")
  //        //              编号(0，1，2)/values/
  //        (x._2, x._1, md5Val) // idx, data, md5_coding
  //      })
  //
  //    var limitCnt = inFileMD5RDD.count().toInt
  //
  //    //limitCnt=limit
  //    if (count.toLong < limitCnt) {
  //      limitCnt = count.toInt
  //    }
  //
  //    //    导入隐式转换
  //    import spark.implicits._
  //
  //    val (colIdx, colData, colMd5) = ("idx", "data", "md5_coding")
  //
  //    // (colIdx, colData, colMd5) = ("idx", "data", "md5_coding")
  //    val inFileMD5DF = inFileMD5RDD.toDF(colIdx, colData, colMd5)
  //
  //    //fieldIndex:
  //    /**
  //      * 返回给定字段值的索引
  //      *
  //      * 当模板未定义时抛出：UnsupportedOperationException
  //      * 当字段名不存在时：throws IllegalArgumentException
  //      *
  //      */
  //
  //
  //    val crossData = inFileMD5DF
  //      .limit(limitCnt)
  //      .join(sqlFullData, colMd5) //?
  //      .map(x => {
  //      (x.getLong(x.fieldIndex(colIdx)), x.getString(x.fieldIndex(colData)), x.getString(x.fieldIndex(colMd5)))
  //    }).toDF(colIdx, colData, colMd5)
  //
  //    val num = crossData.count()
  //
  //    //    crossData.show(numRows = 100)
  //
  //    //    val frame = inFileMD5DF.limit(limitCnt).select(colIdx).except(crossData.select(colIdx))
  //
  //    //    frame.foreach(println)
  //
  //
  //    if (num != limitCnt) {
  //      val diffMd5Rdd = inFileMD5DF.join(
  //        //except 通过对部分行抽样返回一个新的DF:
  //        inFileMD5DF.limit(limitCnt).select(colIdx).except(crossData.select(colIdx)),
  //        colIdx
  //      ).rdd
  //        .sortBy(x => x.getLong(x.fieldIndex(colIdx)))
  //        .map(x => x.getString(x.fieldIndex(colData)))
  //
  //      //      val schema = StructType(colNames.map(fieldName => StructField(fieldName, StringType, true)))
  //      //      val diffMd5Ds = spark.createDataFrame(diffMd5Rdd, schema)
  //      val diffMd5Ds = diffMd5Rdd.toDF()
  //      //      diffMd5Ds.show()
  //      //      diffMd5Ds.toString().foreach(println)
  //
  //      diffMd5Ds.show()
  //
  //      //      val outputDir = "spark/data/result"
  //      //
  //      //      diffMd5Ds.write.mode(SaveMode.Overwrite).text(outputDir.toString)
  //    }

  class Md5Verifier(removeSpace: Boolean = true, delimiter: String = "") extends Serializable {
    var _delimiter: String = delimiter
    var _removeSpace: Boolean = removeSpace

    def md5Verify(content: String): String = {
      _md5Verify(trim(content))
    }

    def trim(content: String): String = {
      var str = content
      if (!_delimiter.isEmpty) {
        str = removeStr(str, _delimiter)
      }
      if (_removeSpace) {
        str = removeStr(str, " ")
      }
      str
    }

    private def _md5Verify(content: String): String = {
      val md5 = MessageDigest.getInstance("MD5")
      val dataMd5 = md5.digest((content).getBytes)
      dataMd5.map("%02x".format(_)).mkString
    }

    def replaceDouble(content: String): String = {
      var str = content
      if (content.indexOf(".") > 0) {
        str = content.replaceAll("0+?$", "").replaceAll("[.]$", "")
      }
      str
    }

    private def removeStr(content: String, removeStr: String): String = {
      return content.replace(removeStr, "")
    }
  }

}
